package com.dongline.test.simple;

import ch.qos.logback.core.util.StatusPrinter;
import com.dongline.test.mapjoin.MapJoinMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Hdfs;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.logging.log4j.core.LoggerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.io.IOException;

public class WordCountDriver {
//    相当于yarn的客户端，需要封装mr运行相关参数，执行jar最后交给yarn
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        System.out.println(WordCountMapper.aa);
        Configuration configuration = new Configuration();
        Logger logger=  LoggerFactory.getLogger(MapJoinMapper.class);
        logger.info("ddddddddddddddddd");
        logger.error("woacao");

        //打印 Logback 内部状态
//        LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
//        StatusPrinter.print(lc);
//        configuration.set("mapreduce.framework.name","yarn");
//        configuration.set("yarn.resourcemanager.hostname","192.168.1.21");

        Job job=Job.getInstance(configuration);





///这句话很重要必须加上
        job.setJarByClass(WordCountDriver.class);


        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);


        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);


        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        ///如果已经存在该路径就删除output文件夹
        FileSystem fileSystem=new Path("/").getFileSystem(configuration);
        if(fileSystem.exists(new Path("D:/output")))
        {
            fileSystem.delete(new Path("D:/output"),true);
        }

        FileInputFormat.setInputPaths(job,new Path("D:/input"));
        FileOutputFormat.setOutputPath(job,new Path("D:/output"));


        boolean re= job.waitForCompletion(true);
//        System.out.println();
        System.exit(re?0:1);


    }
}
